package cn.doitedu.rtmk.demo2;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * @Author: deep as the sea
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/4/14
 * @Desc: 学大数据，到多易教育
 *   Demo2相比demo1来说，规则变得更复杂一点： 可以支持 静态画像条件
 **/
public class RuleEngineCoreFunction2 extends KeyedProcessFunction<Long, UserEvent, JSONObject> {

    Connection hbaseConn;

    @Override
    public void open(Configuration parameters) throws Exception {

        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum","doitedu:2181");
        hbaseConn = ConnectionFactory.createConnection(conf);

    }

    @Override
    public void processElement(UserEvent event, KeyedProcessFunction<Long, UserEvent, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {

        /**
         *  比如规则如下：
         *  静态圈选条件：
         *     月平均消费额(TAG0101) > 200 的用户
         *  触发条件：当他发生A行为时（event_id='page_load', properties['url'] = '/page/001'），命中规则
         */

        // 先判断该行为事件，是否满足规则的触发条件
        if(event.getEvent_id().equals("page_load") && event.getProperties().get("url").equals("/page/001")){
            // 接下来判断，该用户是否是本规则的目标受众：是否满足规则中的画像圈选条件
            Table table = hbaseConn.getTable(TableName.valueOf("user_profile"));

            Get get = new Get(Bytes.toBytes(event.getUser_id()));
            get.addColumn(Bytes.toBytes("f"),Bytes.toBytes("TAG0101"));

            Result result = table.get(get);
            byte[] valueBytes = result.getValue(Bytes.toBytes("f"), Bytes.toBytes("TAG0101"));
            double tagValue = Bytes.toDouble(valueBytes);

            // 判断该用户是否满足本规则的 静态画像条件
            if(tagValue > 200) {

                // 如果满足，则构造结果信息，输出
                JSONObject resultObj = new JSONObject();
                resultObj.put("user_id", event.getUser_id());
                resultObj.put("hit_time",event.getEvent_time());
                resultObj.put("hit_rule","rule001");

                out.collect(resultObj);

            }

        }
    }
}
